package system.config.rocketmq.consumer;

import cn.hutool.json.JSONUtil;
import common.config.rocketmq.MqTaskEnum;
import common.config.rocketmq.RocketMqConstants;
import common.model.dto.ModifyEmailToken;
import common.model.enums.EmailValidEnum;
import common.model.po.SysUser;
import common.util.UserChecker;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.protocol.heartbeat.SubscriptionData;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import system.service.SysUserService;

import javax.annotation.Resource;
import java.util.Objects;

/**
 * @author 米泽鹏
 * @since 2022-5-11 下午 6:08
 */
@Component
@Slf4j
public class ModifyEmailMqConsumer extends DefaultMQPushConsumer implements InitializingBean, DisposableBean {

	@Resource
	private SysUserService userService;
	@Value("${spring.application.name}")
	private String appName;

	@Override
	public void afterPropertiesSet() throws Exception {
		// 初始化时声明消费者组名称
		setConsumerGroup(MqTaskEnum.MODIFY_EMAIL.getConsumerGroup());
		// 指明 nameserver 地址
		setNamesrvAddr(RocketMqConstants.NAMESERVER_ADDRESS);
		// 订阅一个或多个 topic 和 tag（多个tag用||隔开）
		subscribe(MqTaskEnum.MODIFY_EMAIL.getTopic(), SubscriptionData.SUB_ALL);
		// 消息模型：集群模式
		setMessageModel(MqTaskEnum.MODIFY_EMAIL.getMessageModel());
		// 注册消息到达时执行的回调方法
		// 发送2h延时消息，携带 ModifyEmailToken 的信息，到期后，如果用户存在且未删除，且用户此时的邮箱与token中的新邮箱相同，且用户邮箱处于未激活状态，则将邮箱恢复至旧邮箱，且状态设置为已激活
		registerMessageListener((MessageListenerConcurrently) (messages, context) -> {
			try {
				for (MessageExt messageExt : messages) {
					String message = new String(messageExt.getBody());
					ModifyEmailToken modifyEmailToken = JSONUtil.toBean(message, ModifyEmailToken.class);
					Long userId = modifyEmailToken.getUserId();
					String newEmail = modifyEmailToken.getNewEmail();
					String oldEmail = modifyEmailToken.getOldEmail();
					SysUser user = userService.getById(userId);
					UserChecker.checkUser(user);
					if (Objects.equals(user.getEmail(), newEmail) && EmailValidEnum.INVALID.getValue().equals(user.getEmailValid())) {
						user.setEmail(oldEmail);
						user.setEmailValid(EmailValidEnum.VALID.getValue());
						userService.updateById(user);
					}
				}
				log.info(MqTaskEnum.MODIFY_EMAIL.getDescription() + "消息消费成功，消息数量：{}", messages.size());
				return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
			} catch (Exception e) {
				log.error(MqTaskEnum.MODIFY_EMAIL.getDescription() + "消息消费发生异常", e);
				return ConsumeConcurrentlyStatus.RECONSUME_LATER;
			}
		});
		// 消费失败重试次数
		setMaxReconsumeTimes(3);
		// 启动消费者
		start();
		log.info("RocketMQ Consumer启动成功，分组名称：" + MqTaskEnum.MODIFY_EMAIL.getConsumerGroup() + "，来自服务：" + appName);
	}

	@Override
	public void destroy() {
		shutdown();
		log.info("RocketMQ Consumer关闭，分组名称：" + MqTaskEnum.MODIFY_EMAIL.getConsumerGroup() + "，来自服务：" + appName);
	}

}
